//package com.zifeng.kafka;
//
//import com.alibaba.fastjson.JSON;
//import com.zifeng.constant.RedisConstants;
//import com.zifeng.pojo.Block;
//import com.zifeng.pojo.Order;
//import com.zifeng.utils.PowAlgorithm;
//import com.zifeng.utils.RedisUtils;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.annotation.TopicPartition;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//
//import static com.zifeng.constant.RedisConstants.BLOCK_ORDER_PREFIX;
//
//// 改用RocketMQ发送事务消息了，所以用Kafka来消费事务消息
//@Slf4j
//@Component
//public class KafkaConsumer {
//
//    @Autowired
//    private PowAlgorithm powAlgorithm;
//
//    @Autowired
//    private RedisUtils redisUtils;
//
//    @KafkaListener(topics = "order-topic", topicPartitions = {@TopicPartition(topic = "order-topic", partitions = {"0"})})
//    public void processMessage(String content, Acknowledgment acknowledgment) {
//        log.info("收到kafka消息: {}", content);
//        Order order = JSON.parseObject(content, Order.class);
//        // 检查订单是否已存在，做消息消费的幂等处理
//        String orderKey = BLOCK_ORDER_PREFIX + order.getId();
//        String isExist = redisUtils.get(orderKey);
//        if (isExist != null) {
//            log.info("订单已存在，订单ID: {}", order.getId());
//            return;
//        }
//        Block block = powAlgorithm.mine(order);
//        if (block == null) {
//            log.error("挖矿失败，订单ID: {}", order.getId());
//            return;
//        }
//        log.info("创建区块成功，block: {}", block);
//        redisUtils.set(orderKey, "1", RedisConstants.BLOCK_ORDER_TTL, RedisConstants.TTL_TIME_UNIT);
//        // 手动提交offset
//        acknowledgment.acknowledge();
//    }
//
//}